package com.xixi.canal.client.starter.loader;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.xixi.canal.client.DataWriter;
import com.xixi.canal.client.starter.common.SyncSwitch;
import com.xixi.canal.client.starter.config.InstanceConfig;
import com.xixi.canal.client.starter.config.SpringContext;
import com.xixi.canal.client.support.Dml;
import com.xixi.canal.client.support.MessageUtil;
import com.xixi.canal.client.support.ThreadPoolUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public abstract class AbstractCanalClientWorker {

    protected static final Logger logger = LoggerFactory.getLogger(AbstractCanalClientWorker.class);

    protected volatile boolean running = false;

    protected ExecutorService writerExecutors;

    protected Thread thread = null;

    protected Thread.UncaughtExceptionHandler handler = (t , e) -> logger.error("parse events has an error:{}", e);

    protected InstanceConfig instanceConfig;
    // instance
    protected String destination;

    protected List<DataWriter> writers;

    protected SyncSwitch syncSwitch;

    protected ClientIdentity clientIdentity;

    protected CanalServerWithEmbedded embeddedServer;


    public AbstractCanalClientWorker(List<DataWriter> writers) {
        this.writers = writers;
        this.writerExecutors = ThreadPoolUtils.newFixedThreadPool(writers.size(), 5000L);
        syncSwitch = (SyncSwitch)SpringContext.getBean(SyncSwitch.class);

    }

    public void start() {
        if (running) return;
        thread = new Thread(this::process);
        thread.setUncaughtExceptionHandler(handler);
        thread.start();
        running = true;
    }

    public void stop() {

        try {
            if (!running) return;

            running = false;

            writers.forEach(DataWriter::destroy);

            logger.info("## Destination {} all writers destroyed.", destination);


            writerExecutors.shutdown();
            logger.info("## Destination {} writer thread dead.", destination);

            destroy();

            logger.info("## Destination {} is waiting for writer thread die.", destination);

            if (null != thread) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }


    protected void writeMessage(final Message message) {

        List<Future<Boolean>> futures = new ArrayList<>();
        for (DataWriter writer : writers) {
            futures.add(writerExecutors.submit(() -> {
                try {
                    long begin = System.currentTimeMillis();

                    List<Dml> dmlList = MessageUtil.parse4Dml(destination, message);

                    batchSync(writer, dmlList);

                    dmlList = null;
                    if (logger.isDebugEnabled()) {
                        logger.debug("{} elapsed time: {}",
                                writer.getClass().getName(),
                                (System.currentTimeMillis() - begin));
                    }
                    return true;
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                    return false;
                }
            }));
        }
        RuntimeException runtimeException = null;
        for (Future<Boolean> future : futures) {
            try {
                if (!future.get()) {
                    runtimeException = new RuntimeException("Writer sync failed.");
                }
            } catch (Exception e) {
                runtimeException = new RuntimeException(e);
            }
        }
        if (null != runtimeException) throw runtimeException;

    }

    private void batchSync(DataWriter writer, List<Dml> dmlList) {
        if (dmlList.size() < instanceConfig.getSyncBatchSize()) {
            writer.write(dmlList);
        } else {
            List<Dml> batchList = new ArrayList<>();
            for (Dml dml : dmlList) {
                batchList.add(dml);
                int size = batchList.size();
                if (size % instanceConfig.getSyncBatchSize() == 0) {
                    writer.write(batchList);
                    if (null != batchList) {
                        batchList.clear();
                    }
                }
            }

            if (batchList != null && !batchList.isEmpty()) {
                writer.write(batchList);
                batchList.clear();
                batchList = null;
            }
        }
    }

    protected  Message getWithoutAck(int batchSize) {
        Message message = embeddedServer.getWithoutAck(clientIdentity, batchSize);
        Message result = new Message(message.getId());
        if (message.isRaw()) {
            List<ByteString> rawEntries = message.getRawEntries();
            for (ByteString byteString : rawEntries) {
                CanalEntry.Entry entry;
                try {
                    entry = CanalEntry.Entry.parseFrom(byteString);
                } catch (InvalidProtocolBufferException e) {
                    throw new RuntimeException(e);
                }
                result.addEntry(entry);
            }
            result.setRaw(true);
            rawEntries.clear();
            rawEntries = null;
        } else {
            result.setEntries(message.getEntries());
        }
        message = null;
        return result;
    }
    protected void ack(long batchId) {
        embeddedServer.ack(clientIdentity, batchId);
    }
    protected void rollback(long batchId) {
        embeddedServer.rollback(clientIdentity, batchId);
    }
    protected void subscribe() {
        embeddedServer.subscribe(clientIdentity);
    }
    protected void unsubscribe() {
        embeddedServer.unsubscribe(clientIdentity);
    }

    protected abstract void destroy();

    protected abstract void process();

}
